<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="cCXc0" id="cCXc0"><span data-lake-id="uf488f13d" id="uf488f13d">典型回答</span></h1>
  <p data-lake-id="u29f931b4" id="u29f931b4"><br></p>
  <p data-lake-id="ufb22fe2f" id="ufb22fe2f"><span data-lake-id="u2bd4e937" id="u2bd4e937">MQ的消费模式可以大致分为两种，一种是推Push，一种是拉Pull</span></p>
  <p data-lake-id="u260858db" id="u260858db"><br></p>
  <p data-lake-id="u489fd390" id="u489fd390"><span data-lake-id="uc7b45cce" id="uc7b45cce">Push是服务端主动推送消息给客户端，Pull是客户端需要主动到服务端轮询获取数据。</span></p>
  <p data-lake-id="u2b3be63e" id="u2b3be63e"><span data-lake-id="u20faaa9e" id="u20faaa9e">​</span><br></p>
  <p data-lake-id="ub7fc2d3f" id="ub7fc2d3f"><span data-lake-id="u22753290" id="u22753290">他们各自有各自的优缺点，推优点是及时性较好，但如果客户端没有做好流控，一旦服务端推送大量消息到客户端时，就会导致客户端消息堆积甚至崩溃。</span></p>
  <p data-lake-id="uc52dd6ed" id="uc52dd6ed"><span data-lake-id="uabd49dc5" id="uabd49dc5">​</span><br></p>
  <p data-lake-id="u7c59af17" id="u7c59af17"><span data-lake-id="ue19074af" id="ue19074af">拉优点是客户端可以依据自己的消费能力进行消费，但是频繁拉取会给服务端造成压力，并且可能会导致消息消费不及时。</span></p>
  <p data-lake-id="u56938fd0" id="u56938fd0"><span data-lake-id="u6e3b94e2" id="u6e3b94e2">​</span><br></p>
  <p data-lake-id="ue68b2520" id="ue68b2520"><strong><span data-lake-id="u4411967e" id="u4411967e">RocketMQ既提供了Push模式也提供了Pull模式</span></strong><span data-lake-id="ua6df2551" id="ua6df2551">，开发者可以自行选择，主要有两个Consumer可以供开发者选择：</span></p>
  <p data-lake-id="ua4db0ed7" id="ua4db0ed7"><span data-lake-id="ub681f328" id="ub681f328">​</span><br></p>
  <pre lang="java"><code>
public class DefaultMQPullConsumer extends ClientConfig implements MQPullConsumer {

// https://github.com/apache/rocketmq/blob/develop/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
}

public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {

//https://github.com/apache/rocketmq/blob/develop/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
}
</code></pre>
  <p data-lake-id="u9380f9b0" id="u9380f9b0"><br></p>
  <p data-lake-id="u05bb2a5b" id="u05bb2a5b"><span data-lake-id="u60e02cae" id="u60e02cae">其中DefaultMQPullConsumer已经不建议使用了，建议使用DefaultLitePullConsumer。Lite Pull Consumer是RocketMQ 4.6.0推出的Pull Consumer，相比于原始的Pull Consumer更加简单易用，它提供了Subscribe和Assign两种模式。</span></p>
  <p data-lake-id="u57f1b40f" id="u57f1b40f"><span data-lake-id="u374afdc3" id="u374afdc3">​</span><br></p>
  <blockquote data-lake-id="udf1346a9" id="udf1346a9">
   <p data-lake-id="u5b77aa95" id="u5b77aa95"><span data-lake-id="u1f7c30bd" id="u1f7c30bd">/**</span></p>
   <p data-lake-id="u5d788478" id="u5d788478"><span data-lake-id="uee6cb52e" id="uee6cb52e"> * @deprecated Default pulling consumer. This class will be removed in 2022, and a better implementation {@link</span></p>
   <p data-lake-id="uc5d30265" id="uc5d30265"><span data-lake-id="u1bef5e59" id="u1bef5e59"> * DefaultLitePullConsumer} is recommend to use in the scenario of actively pulling messages.</span></p>
   <p data-lake-id="u10e93b2d" id="u10e93b2d"><span data-lake-id="u5948b97e" id="u5948b97e"> */</span></p>
  </blockquote>
  <p data-lake-id="u1886ce4b" id="u1886ce4b"><br></p>
  <p data-lake-id="ueca9ce52" id="ueca9ce52"><span data-lake-id="ud46d4aeb" id="ud46d4aeb">但是，我们需要注意的是，</span><strong><span data-lake-id="ufac67272" id="ufac67272">RocketMQ的push模式其实底层的实现还是基于pull实现的，</span></strong><span data-lake-id="ua73bd03d" id="ua73bd03d">只不过他把pull给封装的比较好，让你以为是在push。</span></p>
  <p data-lake-id="ud20079ef" id="ud20079ef"><strong><span data-lake-id="uf0d1d091" id="uf0d1d091">​</span></strong><br></p>
  <p data-lake-id="u894a0543" id="u894a0543"><span data-lake-id="u0a4cf8fc" id="u0a4cf8fc">​</span><br></p>
  <p data-lake-id="uea604b8a" id="uea604b8a"><span data-lake-id="ude4dcfa6" id="ude4dcfa6">在下面这篇文章中我们介绍过长轮询，其实RocketMQ的push就是通过长轮询来实现的。</span></p>
  <p data-lake-id="u2c51d32a" id="u2c51d32a"><span data-lake-id="u314c4438" id="u314c4438">​</span><br></p>
  <p data-lake-id="u9e43f2be" id="u9e43f2be"><br></p>
  <p data-lake-id="u2810ed32" id="u2810ed32"><span data-lake-id="ubff8f34e" id="ubff8f34e">以下是关于RocketMQ中实现长轮询的代码（基于5.1.4），关键入口PullMessageProcessor的processRequest方法的部分代码：</span></p>
  <p data-lake-id="u23767562" id="u23767562"><span data-lake-id="u8ed7dbd1" id="u8ed7dbd1">​</span><br></p>
  <pre lang="java"><code>
if (this.brokerController.getMessageStore() instanceof DefaultMessageStore) {
    DefaultMessageStore defaultMessageStore = (DefaultMessageStore)this.brokerController.getMessageStore();
    boolean cgNeedColdDataFlowCtr = brokerController.getColdDataCgCtrService().isCgNeedColdDataFlowCtr(requestHeader.getConsumerGroup());
    if (cgNeedColdDataFlowCtr) {
        boolean isMsgLogicCold = defaultMessageStore.getCommitLog()
            .getColdDataCheckService().isMsgInColdArea(requestHeader.getConsumerGroup(),
                requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getQueueOffset());
        if (isMsgLogicCold) {
            ConsumeType consumeType = this.brokerController.getConsumerManager().getConsumerGroupInfo(requestHeader.getConsumerGroup()).getConsumeType();
            if (consumeType == ConsumeType.CONSUME_PASSIVELY) {
                response.setCode(ResponseCode.SYSTEM_BUSY);
                response.setRemark("This consumer group is reading cold data. It has been flow control");
                return response;
            } else if (consumeType == ConsumeType.CONSUME_ACTIVELY) {
                if (brokerAllowFlowCtrSuspend) {  // second arrived, which will not be held
                    PullRequest pullRequest = new PullRequest(request, channel, 1000,
                        this.brokerController.getMessageStore().now(), requestHeader.getQueueOffset(), subscriptionData, messageFilter);
                    this.brokerController.getColdDataPullRequestHoldService().suspendColdDataReadRequest(pullRequest);
                    return null;
                }
                requestHeader.setMaxMsgNums(1);
            }
        }
    }
}
</code></pre>
  <p data-lake-id="u1cdaa86f" id="u1cdaa86f"><br></p>
  <p data-lake-id="uf21955b2" id="uf21955b2"><span data-lake-id="ue20e47ea" id="ue20e47ea">其中这部分代码，就是通过创建一个轮询任务。</span></p>
  <p data-lake-id="u4eea5eef" id="u4eea5eef"><span data-lake-id="u539f3f13" id="u539f3f13">​</span><br></p>
  <pre lang="java"><code>
PullRequest pullRequest = new PullRequest(request, channel, 1000,
                        this.brokerController.getMessageStore().now(), requestHeader.getQueueOffset(), subscriptionData, messageFilter);
this.brokerController.getColdDataPullRequestHoldService().suspendColdDataReadRequest(pullRequest);
</code></pre>
  <p data-lake-id="u3efd097b" id="u3efd097b"><span data-lake-id="u9735454a" id="u9735454a">​</span><br></p>
  <p data-lake-id="ue34addce" id="ue34addce"><a href="https://github.com/apache/rocketmq/blob/develop/broker/src/main/java/org/apache/rocketmq/broker/coldctr/ColdDataPullRequestHoldService.java" target="_blank" data-lake-id="uaea5c888" id="uaea5c888"><span data-lake-id="u2e8b805c" id="u2e8b805c">ColdDataPullRequestHoldService</span></a><span data-lake-id="u486fce2c" id="u486fce2c"> （</span><a href="https://github.com/apache/rocketmq/blob/develop/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PullRequestHoldService.java" target="_blank" data-lake-id="u7b0d4035" id="u7b0d4035"><span data-lake-id="u97c050e7" id="u97c050e7">PullRequestHoldService</span></a><span data-lake-id="u7b634142" id="u7b634142">）是一个子线程，他的run方法如下：</span></p>
  <p data-lake-id="u3d39db86" id="u3d39db86"><span data-lake-id="u343cb5bc" id="u343cb5bc">​</span><br></p>
  <pre lang="java"><code>
@Override
public void run() {
    // 记录服务启动信息
    log.info("{} service started", this.getServiceName());

    // 在服务未停止的情况下循环执行以下逻辑
    while (!this.isStopped()) {
        try {
            // 根据配置决定等待的时长，控制数据流量
            if (!this.brokerController.getMessageStoreConfig().isColdDataFlowControlEnable()) {
                this.waitForRunning(20 * 1000); // 不启用冷数据流量控制时等待 20 秒
            } else {
                this.waitForRunning(5 * 1000);  // 启用冷数据流量控制时等待 5 秒
            }

            // 记录当前时间戳以计算处理时间
            long beginClockTimestamp = this.systemClock.now();

            // 执行检查数据并拉取的逻辑
            this.checkColdDataPullRequest();

            // 计算处理所花费的时间
            long costTime = this.systemClock.now() - beginClockTimestamp;

            // 记录处理耗时，并根据情况标记为 "NOTIFYME" 或 "OK"
            log.info("[{}] checkColdDataPullRequest-cost {} ms.", costTime &gt; 5 * 1000 ? "NOTIFYME" : "OK", costTime);

        } catch (Throwable e) {
            // 记录异常信息，但不中断循环
            log.warn(this.getServiceName() + " service has exception", e);
        }
    }

    // 记录服务结束信息
    log.info("{} service end", this.getServiceName());
}

</code></pre>
  <p data-lake-id="ucdc561ef" id="ucdc561ef"><br></p>
  <p data-lake-id="u6551e6ab" id="u6551e6ab"><span data-lake-id="u292eb9c2" id="u292eb9c2">就是说，每隔一段时间（5秒或者20秒），执行一次数据拉取</span><code data-lake-id="u6fda1135" id="u6fda1135"><span data-lake-id="u2eedf24a" id="u2eedf24a">checkColdDataPullRequest</span></code><span data-lake-id="u47ac6183" id="u47ac6183">，看下这个方法的具体实现：</span></p>
  <pre lang="java"><code>
/**
 * 检查数据并拉取
 */
private void checkColdDataPullRequest() {
    int succTotal = 0, errorTotal = 0;
    int queueSize = pullRequestColdHoldQueue.size();

    // 使用迭代器遍历冷数据拉取请求队列
    Iterator&lt;PullRequest&gt; iterator = pullRequestColdHoldQueue.iterator();
    while (iterator.hasNext()) {
        PullRequest pullRequest = iterator.next();

        // 判断是否超过了冷数据拉取的超时时间
        if (System.currentTimeMillis() &gt;= pullRequest.getSuspendTimestamp() + coldHoldTimeoutMillis) {
            try {
                // 向请求中添加标记表明不需要挂起
                pullRequest.getRequestCommand().addExtField(NO_SUSPEND_KEY, "1");

                // 使用消息处理器执行请求，唤醒客户端进行消息拉取
                this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(
                    pullRequest.getClientChannel(), pullRequest.getRequestCommand());
                succTotal++;
            } catch (Exception e) {
                // 记录异常信息
                log.error("PullRequestColdHoldService checkColdDataPullRequest error", e);
                errorTotal++;
            }

            // 从迭代器中移除已处理的请求
            iterator.remove();
        }
    }

    // 记录处理结果的日志信息
    log.info("checkColdPullRequest-info-finish, queueSize: {} successTotal: {} errorTotal: {}",
        queueSize, succTotal, errorTotal);
}

</code></pre>
  <p data-lake-id="ucd9df871" id="ucd9df871"><span data-lake-id="u216efa2a" id="u216efa2a">​</span><br></p>
  <h1 data-lake-id="dtY4v" id="dtY4v"><span data-lake-id="ufef736f9" id="ufef736f9">扩展知识</span></h1>
  <p data-lake-id="ue84f53c5" id="ue84f53c5"><br></p>
  <h2 data-lake-id="L50ju" id="L50ju"><span data-lake-id="u81bb3887" id="u81bb3887">用法</span></h2>
  <p data-lake-id="u70c0ee90" id="u70c0ee90"><br></p>
  <p data-lake-id="ua0a9f4a2" id="ua0a9f4a2"><span data-lake-id="ue16f6a73" id="ue16f6a73">以下实例来自RocketMQ官网：</span><a href="https://rocketmq.apache.org/" target="_blank" data-lake-id="ued1e3f63" id="ued1e3f63"><span data-lake-id="u8ddcdc39" id="u8ddcdc39">https://rocketmq.apache.org/</span></a></p>
  <h3 data-lake-id="lkaKa" id="lkaKa"><span data-lake-id="u565e8ba3" id="u565e8ba3">Push模式</span></h3>
  <p data-lake-id="u0b894dfe" id="u0b894dfe"><br></p>
  <pre lang="java"><code>
public class Consumer {
  public static void main(String[] args) throws InterruptedException, MQClientException {
    // 初始化consumer，并设置consumer group name
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
   
    // 设置NameServer地址 
    consumer.setNamesrvAddr("localhost:9876");
    //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
    consumer.subscribe("TopicTest", "*");
    //注册回调接口来处理从Broker中收到的消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List&lt;MessageExt&gt; msgs, ConsumeConcurrentlyContext context) {
        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
        // 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    // 启动Consumer
    consumer.start();
    System.out.printf("Consumer Started.%n");
  }
}
</code></pre>
  <p data-lake-id="u8d01c8da" id="u8d01c8da"><br></p>
  <h3 data-lake-id="Jjq4X" id="Jjq4X"><span data-lake-id="uf08e266c" id="uf08e266c">Pull模式</span></h3>
  <p data-lake-id="u7e701647" id="u7e701647"><br></p>
  <pre lang="java"><code>
public class PullConsumerTest {
  public static void main(String[] args) throws MQClientException {
    DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
    consumer.setNamesrvAddr("127.0.0.1:9876");
    consumer.start();
    try {
      MessageQueue mq = new MessageQueue();
      mq.setQueueId(0);
      mq.setTopic("TopicTest");
      mq.setBrokerName("jinrongtong-MacBook-Pro.local");
      long offset = 26;
      PullResult pullResult = consumer.pull(mq, "*", offset, 32);
      if (pullResult.getPullStatus().equals(PullStatus.FOUND)) {
        System.out.printf("%s%n", pullResult.getMsgFoundList());
        consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
    consumer.shutdown();
  }
}
</code></pre>
  <p data-lake-id="uab98561b" id="uab98561b"><br></p>
  <p data-lake-id="u25d63db7" id="u25d63db7"><br></p>
  <pre lang="java"><code>
public class LitePullConsumerSubscribe {
    public static volatile boolean running = true;
    public static void main(String[] args) throws Exception {
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("lite_pull_consumer_test");
        litePullConsumer.subscribe("TopicTest", "*");
        litePullConsumer.setPullBatchSize(20);
        litePullConsumer.start();
        try {
            while (running) {
                List&lt;MessageExt&gt; messageExts = litePullConsumer.poll();
                System.out.printf("%s%n", messageExts);
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
}
</code></pre>
 </body>
</html>